package cn.lsh.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class TimeProducerInterceptor implements ProducerInterceptor<Integer, String> {

	@Override
	public ProducerRecord<Integer, String> onSend(ProducerRecord<Integer, String> producerRecord) {
		//消息发送前调用
		return new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(),
				producerRecord.key(), producerRecord.value() + "," + System.currentTimeMillis(),
				producerRecord.headers());
	}

	@Override
	public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
		//消息发送到kafka之后调用或在发送过程中失败是调用
	}

	@Override
	public void close() {

	}

	@Override
	public void configure(Map<String, ?> map) {

	}
}
